# Copyright (c) HySoP 2011-2024
#
# This file is part of HySoP software.
# See "https://particle_methods.gricad-pages.univ-grenoble-alpes.fr/hysop-doc/"
# for further info.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from abc import ABCMeta, abstractmethod
from hysop.tools.htypes import check_instance
from hysop.topology.topology import Topology, TopologyView
from hysop.mesh.mesh import Mesh
[docs]
class TopologyDescriptor(metaclass=ABCMeta):
"""
Describes how a topology should be built.
Multiple compatible topology descriptors are gathered during
operator graph building and are replaced by a single unique
topology upon initialization.
"""
__slots__ = ("_mpi_params", "_domain", "_backend", "_extra_kwds")
def __init__(self, mpi_params, domain, backend, **kwds):
"""
Initialize a TopologyDescriptor.
Notes
-----
kwds allows for backend specific variables.
CartesianTopologyDescriptor is immutable.
"""
super().__init__()
self._mpi_params = mpi_params
self._domain = domain
self._backend = backend
self._extra_kwds = frozenset(kwds.items())
if ("cl_env" in kwds) and (kwds["cl_env"] is not None):
assert kwds["cl_env"].mpi_params is mpi_params
def _get_mpi_params(self):
"""Get mpi parameters."""
return self._mpi_params
def _get_domain(self):
"""Get domain."""
return self._domain
def _get_backend(self):
"""Get backend."""
return self._backend
def _get_dim(self):
"""Get domain dimension."""
return self.domain.dim
def _get_extra_kwds(self):
"""Get extra keyword arguments."""
return dict(self._extra_kwds)
mpi_params = property(_get_mpi_params)
domain = property(_get_domain)
backend = property(_get_backend)
extra_kwds = property(_get_extra_kwds)
dim = property(_get_dim)
[docs]
@staticmethod
def build_descriptor(backend, operator, field, handle, **kwds):
"""
Generate a descriptor from a lower level representation.
If handle is already a Topology or a TopologyDescriptor it
is returned unchanged.
If handle is a CartesianTopologyDescriptors (ie. currently a Discretization),
a CartesianTopologyDescriptor is created and returned.
Every new topology type should be registered here.
"""
from hysop.topology.cartesian_descriptor import (
CartesianTopologyDescriptor,
CartesianTopologyDescriptors,
)
if isinstance(handle, Topology):
# handle is already a Topology, so we return it.
return handle
elif isinstance(handle, TopologyView):
return handle._topology
elif isinstance(handle, TopologyDescriptor):
# handle is already a TopologyDescriptor, so we return it.
return handle
elif isinstance(handle, CartesianTopologyDescriptors):
return CartesianTopologyDescriptor.build_descriptor(
backend, operator, field, handle, **kwds
)
elif handle is None:
# this topology will be determined later
return None
else:
msg = "Unknown handle of class {} to build a TopologyDescriptor."
msg = msg.format(handle.__class__)
raise TypeError(msg)
[docs]
def choose_or_create_topology(self, known_topologies, **kwds):
"""
Returns a topology that is either taken from known_topologies, a set
of user specified topologies which are ensured to be compatible
with the current TopologyDescriptor, or created from the descriptor
if choose_topology() returns None.
"""
check_instance(known_topologies, set, values=Topology)
topo = self.choose_topology(known_topologies, **kwds)
if topo is None:
topo = self.create_topology(**kwds)
return topo
[docs]
@abstractmethod
def choose_topology(self, known_topologies, **kwds):
"""
Find optimal topology parameters from known_topologies.
If None is returned, create_topology will be called instead.
"""
pass
[docs]
@abstractmethod
def create_topology(self, **kwds):
"""
Build a topology with the current TopologyDescriptor.
"""
pass
[docs]
def match(self, other, invert=False):
"""Test if this descriptor is equivalent to the other one."""
if not isinstance(other, TopologyDescriptor):
return NotImplemented
eq = self.domain is other.domain
eq &= self.mpi_params == other.mpi_params
eq &= self.backend == other.backend
eq &= self.extra_kwds == other.extra_kwds
if invert:
return not eq
else:
return eq
@abstractmethod
def __eq__(self, other):
return self.match(other)
@abstractmethod
def __ne__(self, other):
return not self.match(other, invert=True)
@abstractmethod
def __hash__(self):
h = id(self.domain)
h ^= hash(self.mpi_params)
h ^= hash(self.backend)
h ^= hash(self._extra_kwds)
return h
@abstractmethod
def __str__(self):
pass
TopologyDescriptors = (Topology, TopologyDescriptor, type(None))
"""
Instance of those types can be used to create a TopologyDescriptor.
Thus they can be passed in the variables of each operator.
"""